package com.atguigu.realtime.utils;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Properties;

/**
 * Created by Smexy on 2022/8/24
 */
public class KafkaProducerUtil
{

    //有生产者
    private static Producer<String,String> producer;

    //为Producer赋值
    static {

        //放生产者的配置
        Properties properties = new Properties();

        /*
            必须配置的是
                    集群地址
                    Key,value的序列化器

            配置可以从 ProducerConfig中查阅
         */
        properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,PropertiesUtil.getValue("kafka.broker.list"));
        properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,PropertiesUtil.getValue("key.serializer"));
        properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,PropertiesUtil.getValue("value.serializer"));

        //设置生产者是一个幂等生产者
        properties.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG,"true");

        producer = new KafkaProducer<String,String>(properties);

    }

    //提供一个生产数据的方法
    public static void sendData(String data,String topic){

        //用生产者去生产
        producer.send(new ProducerRecord<String,String>(topic,data));

    }

    public static void flush(){
        producer.flush();
    }
}
